package org.groupg.project;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.subjects.PublishSubject;
import io.reactivex.rxjava3.subscribers.ResourceSubscriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

public class FixedRxJavaDisruptorETL {
    static Logger log = LoggerFactory.getLogger(FixedRxJavaDisruptorETL.class);

    // 1. 定义ETL事件类
    static class EtlEvent {
        private long id;
        private String rawData;
        private String transformedData;
        private boolean processed;
        private Throwable error;

        public void reset() {
            id = 0;
            rawData = null;
            transformedData = null;
            processed = false;
            error = null;
        }

        @Override
        public String toString() {
            return "EtlEvent{id=" + id + ", rawData='" + rawData + "'}";
        }
    }

    // 2. 事件工厂
    static class EtlEventFactory implements EventFactory<EtlEvent> {
        @Override
        public EtlEvent newInstance() {
            return new EtlEvent();
        }
    }

    // 3. 事件处理器 - 数据转换
    static class TransformationHandler implements EventHandler<EtlEvent> {
        private final DebugUtils.EventTracker tracker;

        public TransformationHandler(DebugUtils.EventTracker tracker) {
            this.tracker = tracker;
        }

        @Override
        public void onEvent(EtlEvent event, long sequence, boolean endOfBatch) {
            try {
                log.info("[Transformation] 开始处理: " + event.id);

                // 模拟数据转换逻辑
                event.transformedData = event.rawData.toUpperCase() + "_PROCESSED";
                event.processed = true;

                // 模拟处理延迟
                TimeUnit.MILLISECONDS.sleep(1);

                tracker.eventTransformed();
                log.info("[Transformation] 完成处理: " + event.id);
            } catch (Exception e) {
                event.error = e;
                log.error("[Transformation] 处理错误: " + e.getMessage());
            }
        }
    }

    // 4. 修复后的背压控制器
    static class FixedBackpressureController implements EventHandler<EtlEvent> {
        private final PublishSubject<EtlEvent> subject;
        private final AtomicLong requested = new AtomicLong(0);
        private final DebugUtils.EventTracker tracker;

        // 添加超时机制
        private final long maxWaitTimeMs = 1000;

        public FixedBackpressureController(PublishSubject<EtlEvent> subject,
                                           DebugUtils.EventTracker tracker) {
            this.subject = subject;
            this.tracker = tracker;
        }

        @Override
        public void onEvent(EtlEvent event, long sequence, boolean endOfBatch) {
            log.info("[Backpressure] 收到事件: " + event.id);

            long startTime = System.currentTimeMillis();
            boolean requestSatisfied = false;

            try {
                // 检查是否有请求（带超时）
                while (requested.get() <= 0) {
                    if (System.currentTimeMillis() - startTime > maxWaitTimeMs) {
                        log.info("[Backpressure] 等待请求超时: " + event.id);
                        throw new TimeoutException("背压请求等待超时");
                    }

                    // 短暂等待避免忙循环
                    TimeUnit.MILLISECONDS.sleep(10);
                }

                // 发送事件到RxJava流水线
                subject.onNext(event);
                requested.decrementAndGet();
                tracker.eventPassedToRx();
                requestSatisfied = true;
                log.info("[Backpressure] 发送事件到RxJava: " + event.id);
            } catch (Exception e) {
                log.error("[Backpressure] 处理错误: " + e.getMessage());
                event.error = e;
            } finally {
                if (!requestSatisfied) {
                    log.info("[Backpressure] 事件未发送: " + event.id);
                }
            }
        }

        public void request(long n) {
            log.info("[Backpressure] 收到RxJava请求: " + n);
            requested.addAndGet(n);
        }
    }

    public static void main(String[] args) throws InterruptedException, TimeoutException, com.lmax.disruptor.TimeoutException {
        // 启动死锁检测
        DebugUtils.deadlockDetector();

        // 创建事件追踪器
        DebugUtils.EventTracker tracker = new DebugUtils.EventTracker();

        // 配置参数
        final int BUFFER_SIZE = 16; // 小缓冲区便于观察
        final int PRODUCER_COUNT = 2;
        final int RX_BUFFER_SIZE = 8;

        // 5. 创建Disruptor（使用更快的等待策略）
        Disruptor<EtlEvent> disruptor = new Disruptor<>(
                new EtlEventFactory(),
                BUFFER_SIZE,
                DaemonThreadFactory.INSTANCE,
                ProducerType.MULTI,
                new YieldingWaitStrategy() // 更高效的等待策略
        );

        // 6. 设置异常处理器
        disruptor.setDefaultExceptionHandler(new ExceptionHandler<EtlEvent>() {
            @Override
            public void handleEventException(Throwable ex, long sequence, EtlEvent event) {
                log.error("Disruptor事件异常: " + ex.getMessage());
            }

            @Override
            public void handleOnStartException(Throwable ex) {
                log.error("Disruptor启动异常: " + ex.getMessage());
            }

            @Override
            public void handleOnShutdownException(Throwable ex) {
                log.error("Disruptor关闭异常: " + ex.getMessage());
            }
        });

        // 7. 创建RxJava背压控制器
        PublishSubject<EtlEvent> rxSubject = PublishSubject.create();
        FixedBackpressureController backpressureController =
            new FixedBackpressureController(rxSubject, tracker);

        // 8. 设置Disruptor处理器链（添加追踪器）
        disruptor.handleEventsWith(new TransformationHandler(tracker))
                .then(backpressureController);

        // 9. 启动Disruptor
        RingBuffer<EtlEvent> ringBuffer = disruptor.start();
        log.info("Disruptor启动完成，缓冲区大小: " + BUFFER_SIZE);

        // 10. 创建RxJava处理流水线
        Flowable<EtlEvent> etlPipeline = rxSubject.toFlowable(BackpressureStrategy.BUFFER)
                .onBackpressureBuffer(RX_BUFFER_SIZE,
                        () -> log.info("Rx Buffer Overflow!"),
                        BackpressureOverflowStrategy.DROP_LATEST)
                .observeOn(Schedulers.io(), false, RX_BUFFER_SIZE)
                .doOnNext(event -> {
                    log.info("[RxJava] 开始处理: " + event.id);
                    try {
                        if (event.error != null) {
                            log.error("[RxJava] 处理错误: " + event.error.getMessage());
                        } else if (event.processed) {
                            // 模拟数据加载到目标系统
                            TimeUnit.MILLISECONDS.sleep(10);
                            tracker.eventConsumed();
                            log.info("[RxJava] 完成处理: " + event.id);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });

        // 11. 注册背压请求回调
        ResourceSubscriber<EtlEvent> subscriber = new ResourceSubscriber<EtlEvent>() {
            @Override
            protected void onStart() {
                log.info("[RxJava] 订阅开始，请求初始数据");
                request(1);
            }

            @Override
            public void onNext(EtlEvent event) {
                log.info("[RxJava] 接收事件: " + event.id);

                // 请求下一个事件
                request(1);
            }

            @Override
            public void onError(Throwable t) {
                log.error("[RxJava] 流水线错误: " + t.getMessage());
            }

            @Override
            public void onComplete() {
                log.info("[RxJava] ETL处理完成");
            }
        };

        etlPipeline.subscribe(subscriber);

        // 12. 创建生产者线程池
        ExecutorService producerPool = Executors.newFixedThreadPool(PRODUCER_COUNT, r -> {
            Thread t = new Thread(r, "Producer-" + PRODUCER_COUNT);
            t.setDaemon(true);
            return t;
        });

        AtomicLong eventCounter = new AtomicLong(1);

        // 13. 启动生产者（添加详细日志）
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            final int producerId = i + 1;
            producerPool.submit(() -> {
                try {
                    log.info("生产者 " + producerId + " 启动");
                    while (!Thread.currentThread().isInterrupted()) {
                        // 申请序列号（带超时）
                        long sequence = ringBuffer.tryNext();
                        if (sequence == -1) {
                            log.info("生产者 " + producerId + " 缓冲区满，等待...");
                            TimeUnit.MILLISECONDS.sleep(100);
                            continue;
                        }

                        EtlEvent event = ringBuffer.get(sequence);

                        // 重置事件对象
                        event.reset();

                        // 设置事件数据
                        long eventId = eventCounter.getAndIncrement();
                        event.id = eventId;
                        event.rawData = "原始数据_" + eventId;

                        tracker.eventProduced();
                        log.info("生产者 " + producerId + " 发布事件: " + eventId);

                        // 发布事件
                        ringBuffer.publish(sequence);

                        // 控制生产速率
                        TimeUnit.MILLISECONDS.sleep(50);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.info("生产者 " + producerId + " 被中断");
                } catch (Exception e) {
                    log.error("生产者 " + producerId + " 错误: " + e.getMessage());
                }
            });
        }

        // 14. 运行监控
        ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
        monitor.scheduleAtFixedRate(() -> {
            log.info("\n===== 系统状态监控 =====");
            log.info("生产事件总数: " + (eventCounter.get() - 1));
            log.info("Disruptor剩余容量: " + ringBuffer.remainingCapacity());
            log.info("背压控制器请求数: " + backpressureController.requested.get());
            tracker.printStats();

            // 打印线程状态
            log.info("\n===== 活动线程 =====");
            Thread.getAllStackTraces().keySet().forEach(t ->
                log.info(t.getName() + " - " + t.getState()));

        }, 3, 3, TimeUnit.SECONDS);

        // 15. 运行一段时间后停止
        TimeUnit.SECONDS.sleep(30);

        log.info("===== 停止系统 =====");

        // 16. 关闭资源
        producerPool.shutdownNow();
        if (!producerPool.awaitTermination(3, TimeUnit.SECONDS)) {
            log.error("生产者线程未正常终止");
        }

        disruptor.shutdown(5, TimeUnit.SECONDS);
        subscriber.dispose();
        rxSubject.onComplete();
        monitor.shutdownNow();

        // 最终状态报告
        log.info("\n===== 最终状态报告 =====");
        tracker.printStats();
        log.info("ETL系统已关闭");
    }
}
